package reactor.core.publisher;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;

public class FluxDemo {

    private static final Logger log = LoggerFactory.getLogger(FluxDemo.class);

    public static Flux<Object> baseFlowable() {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        return
            Flux
                .create(emitter -> {
                    emitter(emitter, atomicInteger);
                    CompletableFuture<Void> future0 = CompletableFuture.runAsync(() -> emitter(emitter, atomicInteger));
                    emitter(emitter, atomicInteger);
                    CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> emitter(emitter, atomicInteger));
                    CompletableFuture.allOf(future0, future1).thenRun(emitter::complete);
                });
    }

    private static void emitter(FluxSink<Object> emitter, AtomicInteger atomicInteger) {
        int i = atomicInteger.addAndGet(1);
        log.info("输出:{}", i);
        emitter.next(i);
    }

    public static void subscribeFour(Flux<Object> flowable) {
        subscribeFour(flowable, new AtomicInteger(1));
    }

    public static void subscribeFour(Flux<Object> flowable, AtomicInteger index) {
        baseSubscribe(flowable, index);
        baseSubscribe(flowable, index);

        final int indexNum = index.getAndAdd(1);
        log.info("第{}次订阅", indexNum);
        flowable.subscribe(o -> {
            log.info("第{}次消费并报错：{}", indexNum, o);
            throw new RuntimeException("主动报错");
        }, e -> log.error("第{}次消费出现异常:{}", indexNum, e.getMessage()), () -> log.info("第{}次订阅完成事件", indexNum));
        baseSubscribe(flowable, index);
    }

    private static void baseSubscribe(Flux<Object> flowable, AtomicInteger index) {
        final int indexNum = index.getAndAdd(1);
        log.info("第{}次订阅", indexNum);
        flowable.subscribe(o -> logSubscribe(indexNum, o), null, () -> log.info("第{}次订阅完成事件", indexNum));
    }

    private static void logSubscribe(int index, Object o) {
        log.info("第{}次消费：{}", index, o);
    }

}
